package com.zhang.spark_streaming.status

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @title:
 * @author: zhang
 * @date: 2021/12/13 08:42 
 */
object SparkStreaming_state {

  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("sparkStreaming")
    val ssc = new StreamingContext(sparkConf,Seconds(3))

    ssc.checkpoint("cp")

    val line: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)

    val wordToOne: DStream[(String, Int)] = line.flatMap(_.split(" ")).map((_, 1))

    //updateStateByKey:根据key对数据的状态进行更新
    //第一个参数表示相同的key的value
    //第二个参数表示缓存区相同key的value数据
    val wordToCount: DStream[(String, Int)] = wordToOne.updateStateByKey(
      (seq: Seq[Int], buff: Option[Int]) => {
        val newCount = buff.getOrElse(0) + seq.sum
        Option(newCount)
      }
    )
    wordToCount.print()

    ssc.start()
    ssc.awaitTermination()
  }

}
